package com.example.czy.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 多线程工具
 *
 * @Author czy
 * @Date 2023-12-26 22:29:52
 **/
@Slf4j
public class ThreadUti {

    /**
     * 分批执行示例 比如5000条数据
     * 先执行当前页数据，执行成功返回结果，其他页数据多线程分批执行
     *
     * @param args
     */
    public static void main(String[] args) {
        List list = new ArrayList();
        for (int i = 1; i <= 10000; i++) {
            list.add(i);
        }
        executeBatches(list, 1000);
        // threadedSharing();
        // executeBatches2(list);


    }


    /**
     * 线程变量共享
     */
    private static void threadedSharing() {
        InheritableThreadLocal<Object> param = new InheritableThreadLocal();
        param.set("data");

        ThreadPoolTaskExecutor threadPoolTaskExecutor = getThreadPoolTaskExecutor();

        CompletableFuture.runAsync(() -> System.out.println("data:" + param.get()), threadPoolTaskExecutor);
    }

    /**
     * 多线程分批执行
     * 传入集合、批次数据 分批执行
     *
     * @param list      集合
     * @param batchSize 批次数量
     */
    private static void executeBatches(List<?> list, Integer batchSize) {

        // 当前页数据
        List<?> page = CommUtils.getPage(list, 1, 1000);

        // 执行当前页数据
        sync(page, batchSize);

        // 去除当前页数据，其他页数据 多线程分批执行
        list.removeAll(page);
        List<CompletableFuture<Boolean>> sync = sync(list, batchSize);

        // 等待所有任务完成
        CompletableFuture<Void> allTasks = CompletableFuture.allOf(sync.toArray(new CompletableFuture[0]));
        allTasks.thenRun(() -> {
            log.info("thread task execution completed!");
            list.clear();
        }).exceptionally(ex -> {
            log.error("thread task execution error：{}", ex.getMessage());
            // 执行清理操作
            list.clear();
            return null;
        });

    }

    public static List<CompletableFuture<Boolean>> sync(List<?> list, Integer batchSize) {
        // 线程数
        int threadNum = (list.size() + batchSize - 1) / batchSize;
        if (threadNum > 10) {
            threadNum = 10;
            batchSize = (int) Math.ceil((double) list.size() / threadNum);
        }


        List<CompletableFuture<Boolean>> tasks = new ArrayList<>();
        for (int i = 0; i < threadNum; i++) {
            try {
                Thread.sleep(200L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            int startIndex = i * batchSize;
            int endIndex = Math.min((i + 1) * batchSize, list.size());
            List longs = list.subList(startIndex, endIndex);
            tasks.add(CompletableFuture.supplyAsync(() -> {
                log.info(longs.toString());
                return true;
            }));
        }
        return tasks;
    }


    /**
     * 初始化线程池
     *
     * @return
     */
    public static ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        // 核心线程大小，线程池中维护一个最小的线程数量，即使这些线程处于空闲状态，也一直存在池中，除非设置了核心线程超时时间
        threadPoolTaskExecutor.setCorePoolSize(20);
        // 线程池中允许的最大线程数量
        threadPoolTaskExecutor.setMaxPoolSize(100);
        // 任务队列容量（阻塞队列）
        threadPoolTaskExecutor.setQueueCapacity(200);
        threadPoolTaskExecutor.setKeepAliveSeconds(10);
        threadPoolTaskExecutor.setThreadNamePrefix("Async-Service-");
        // 线程池对拒绝任务的处理策略
        // CallerRunsPolicy:由调用线程（提交任务的线程）处理该任务
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}
